package com.hzgj.soa.spring;

import com.google.common.collect.Maps;
import com.google.common.collect.Sets;
import com.hzgj.bcl.soa.ChainStack;
import com.hzgj.bcl.soa.Event;
import com.hzgj.bcl.soa.Interceptor;
import com.hzgj.bcl.soa.cicada.Cicada;
import com.hzgj.bcl.soa.cicada.lb.BaseLoadBalancer;
import com.hzgj.bcl.soa.cicada.lb.Rule;
import com.hzgj.bcl.soa.cicada.pool.ChannelPoolConfig;
import com.hzgj.bcl.soa.cicada.router.CicadaRouter;
import com.hzgj.bcl.soa.config.ChannelConfig;
import com.hzgj.bcl.soa.config.ReferenceDef;
import com.hzgj.bcl.soa.interceptor.SerializationInterceptor;
import com.hzgj.bcl.soa.registry.RegistryService;
import com.hzgj.bcl.soa.rpc.RpcDispatcher;
import com.hzgj.bcl.soa.rpc.ServiceAddress;
import com.hzgj.bcl.soa.util.Constants;
import com.hzgj.bcl.soa.util.ServiceLoaderUtil;
import com.hzgj.bcl.util.jmx.Management;
import com.hzgj.bcl.util.lang.StrKit;
import lombok.Setter;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.FactoryBean;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;

import java.util.Map;
import java.util.Set;
import java.util.stream.Collectors;

/**
 * Created by warning5 on 2017/3/7.
 */
@Setter
public class RpcDispatcherBean implements ApplicationContextAware, FactoryBean {
    private Logger logger = LoggerFactory.getLogger(getClass());
    private ApplicationContext applicationContext;
    private ChainStack chainStack = new ChainStack();

    public void handleReferences() {

        RegistryBean registryBean = null;
        if (applicationContext.containsBean(RegistryBean.class.getSimpleName())) {
            registryBean = applicationContext.getBean(RegistryBean.class);
        }

        Map<String, ReferenceBean> referenceDefs = applicationContext.getBeansOfType(ReferenceBean.class);
        if (referenceDefs.size() == 0) {
            if (logger.isDebugEnabled()) {
                logger.debug("no reference,it is server without monitor.");
            }
        }
        RegistryService registryService = null;
        if (registryBean != null) {
            registryService = registryBean.getRegistryService();
        }

        Map<ReferenceDef, ServiceAddress> serviceAddresses = Maps.newHashMap();
        Map<String, ChannelConfig> channelConfigs = Maps.newHashMap();
        Map<String, ChannelPoolConfig> channelPoolConfigs = applicationContext.getBeansOfType(ChannelPoolConfig.class);
        Map<String, Rule> rules = Maps.newHashMap();
        Set<String> ruleNames = Sets.newHashSet();
        for (ReferenceBean referenceBean : referenceDefs.values()) {
            ReferenceDef referenceDef = referenceBean.getReferenceDef();
            if (registryBean == null && StrKit.isBlank(referenceDef.getUrl())) {
                throw new RuntimeException("neither url or registry not exist for " + referenceDef);
            }
            //包含所有的服务引用
            if (referenceDef.isCheck()) {
                serviceAddresses.put(referenceDef, ServiceAddress.getAddress(referenceDef, registryService));
            } else {
                serviceAddresses.put(referenceDef, null);
            }
            ChannelConfig channelConfig = channelConfigs.get(referenceDef.getProtocol());
            if (channelConfig == null) {
                channelConfig = new ChannelConfig();
                channelConfig.setProtocol(referenceDef.getProtocol());
                channelConfigs.put(referenceDef.getProtocol(), channelConfig);
            }

            ruleNames.add(referenceDef.getLoadbalance());
            if (referenceDef.getMethodDefs() != null) {
                ruleNames.addAll(referenceDef.getMethodDefs().values().stream().filter(methodDef -> StrKit.notBlank(methodDef.getLoadbalance())).map(methodDef ->
                        methodDef.getLoadbalance()).collect(Collectors.toSet()));
            }
        }
        ServiceLoaderUtil.getServiceLoaderObjects(Rule.class).forEachRemaining(rule -> {
            if (ruleNames.contains(rule.getName())) {
                rules.put(rule.getName(), rule);
            }
        });

        //TODO 只添加了序列化拦截器，并且无法配置
        chainStack.addInterceptor(new SerializationInterceptor());
        RegistryService _registryService = registryService;
        chainStack.insertInterceptorAtBottom(new Interceptor() {
            Cicada cicada = null;

            @Override
            public void start() throws Exception {
                super.start();
                CicadaRouter cicadaRouter = new CicadaRouter(serviceAddresses, channelPoolConfigs, channelConfigs, _registryService);
                Management.register(cicadaRouter, Constants.getCicadaRouterObjectName());
                cicada = new Cicada(new BaseLoadBalancer(rules, cicadaRouter), event -> this.up(event));
                cicada.start();
            }

            @Override
            public Object down(Event evt) {
                return cicada.send(evt);
            }

            @Override
            public void stop() {
                if (cicada != null) {
                    cicada.stop();
                }
                super.stop();
            }
        });
    }

    @Override
    public Object getObject() throws Exception {
        handleReferences();
        return new RpcDispatcher(chainStack);
    }

    @Override
    public Class<?> getObjectType() {
        return RpcDispatcher.class;
    }

    @Override
    public boolean isSingleton() {
        return true;
    }
}
